package com.xuhai

import com.alibaba.fastjson.JSON
import com.xuhai.entity.{GPSDATA, ResultBean}
import com.xuhai.util.{HttpClientUtils, IsPointInPolygon}
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.JavaConversions._
import scala.collection.mutable.ListBuffer

/**
  * 计算进场率
  */
object EntryRate {

  val path = "file:///D:\\entryRate20180618.txt"

  def main(args: Array[String]): Unit = {

    val sparkConf = new SparkConf().setAppName("entryRate").setMaster("local[4]")
    val sc = new SparkContext(sparkConf)
    val trdd = sc.textFile(path)

    //获取场站信息
    val resultStr = HttpClientUtils.get("http://172.19.66.31:8080/gj/gj/exam/carEntryRateAction.do?action=getStationInfo")
    val cityJSONArray = JSON.parseObject(resultStr).getJSONArray("result")
    val citylen = cityJSONArray.size()

    val resultList = new java.util.ArrayList[ResultBean]()

    for (i <- 0 until citylen){
      //按每个城市统计进场率
      val cityJSONObject =  cityJSONArray.getJSONObject(i)
      var city = cityJSONObject.getString("city")
      if(city != null){
        city = if(city.equals("520300")) "520301" else city
        val stationJSONArray = cityJSONObject.getJSONArray("stations")
        val stationlen = stationJSONArray.size()
        //收集所有场站
        val stationPloygonList = new ListBuffer[String]
        for (i <- 0 until stationlen) {
          val stationJSONObject = stationJSONArray.getJSONObject(i)
          if(stationJSONObject.getJSONArray("rings") != null && stationJSONObject.getJSONArray("rings").size() > 0){
            stationPloygonList.add(stationJSONObject.getJSONArray("rings").getJSONArray(0).toString)
          }
        }

        val rdd = trdd.filter(str => str.startsWith(city)).map(str => {
          val arr = str.split(",")
          GPSDATA(arr(0),arr(1),arr(2),arr(3),arr(4),arr(5))
        })
        //城市在线车辆总数
        val total = rdd.count()
        //进场的车辆总数
        val enterTotal = rdd.filter(gpsData => isInPolygon(gpsData, stationPloygonList)).count()
        if(total > 0){
          val rate = (BigDecimal(enterTotal)/BigDecimal(total)).setScale(2, BigDecimal.RoundingMode.HALF_UP)
          val result = new ResultBean()
          result.setCity(city)
          result.setRate(rate.bigDecimal.toString)
          resultList.add(result)
        }
      }
    }

    println(JSON.toJSON(resultList))
  }

  def isInPolygon(gpsData: GPSDATA, stationPloygonList: ListBuffer[String]): Boolean = {
    for(stationPloygon <- stationPloygonList){
      if(IsPointInPolygon.isInPolygon(IsPointInPolygon.transferToDouble2Array(stationPloygon), BigDecimal(gpsData.gpsLon).doubleValue(),BigDecimal(gpsData.gpsLat).doubleValue())){
        return true
      }
    }
    return false
  }

}
